package com.example.mq.mqserver.datacenter;

import com.example.mq.common.MQException;
import com.example.mq.mqserver.core.Binding;
import com.example.mq.mqserver.core.Exchange;
import com.example.mq.mqserver.core.MQueue;
import com.example.mq.mqserver.core.Message;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 这个类统一管理内存中的所有数据
 */
@Slf4j
@Data
public class MemoryDataCenter {
    // 1. 交换机: key 是 exchangeName，value 是 Exchange 对象
    private final ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    // 2. 队列: key 是 queueName，value 是 MQueue 对象
    private final ConcurrentHashMap<String, MQueue> queueMap = new ConcurrentHashMap<>();
    // 3. 绑定(交换机和队列之间的关联): key 是 exchangeName，value 是哈希表（key 是 queueName，value 是 Binding 对象）
    private final ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    // 4. 消息: key 是 messageId，value 是 Message 对象
    private final ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    // 5. 队列和消息的关联: key 是 queueName，value 是存放 Message 对象的链表
    private final ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    // 5. 队列和待确认消息的关联: key 是 queueName，value 是哈希表（key 是 messageId，value 是 Message 对象）
    private final ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    // 交换机管理
    // 1. 新增交换机
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        log.info("新增交换机成功！exchangeName=" + exchange.getName());
    }
    // 2. 获取交换机
    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }
    // 3. 删除交换机
    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        log.info("交换机删除成功！exchangeName=" + exchangeName);
    }

    // 队列管理
    // 1. 新增队列
    public void insertQueue(MQueue queue) {
        queueMap.put(queue.getName(), queue);
        log.info("新增队列成功！queueName=" + queue.getName());
    }
    // 2. 获取队列
    public MQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }
    // 3. 删除队列
    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        log.info("队列删除成功！queueName=" + queueName);
    }

    // 绑定管理
    // 1. 新增绑定
    public void insertBinding(Binding binding) throws MQException {
        // 1. 先使用 exchangeName 查找，看一下对应的哈希表是否存在，不存在则创建一个
        ConcurrentHashMap<String, Binding> bindingMap
                = bindingsMap.computeIfAbsent(binding.getExchangeName(), k -> new ConcurrentHashMap<>());

        // 2. 再根据 queueName 查找，如果不存在，就插入，如果已经存在，就抛出异常
        // 需要加锁，这里是两步操作，第一步判断是否存在，第二步插入
        // 使用当前操作对象 bindingMap 进行加锁
        synchronized (bindingMap) {
            if (bindingMap.get(binding.getQueueName()) != null) {
                throw new MQException("[MemoryDataCenter] 绑定已经存在！exchangeName="
                        + binding.getExchangeName() + ", queueName=" + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
        }
        log.info("新增绑定成功！exchangeName="
                + binding.getExchangeName() + ", queueName=" + binding.getQueueName());
    }
    // 2. 获取绑定，两个版本
    // 1) 根据 exchangeName 和 queueName 确定唯一一个 Binding
    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
    }
    // 2) 根据 exchangeName 获取到交换机所有的 Binding
    // 返回一个交换机 对应的 绑定表
    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }
    // 3. 删除绑定
    public void deleteBinding(Binding binding) throws MQException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            throw new MQException("[MemoryDataCenter] 绑定不存在！exchangeName="
                    + binding.getExchangeName() + ", queueName=" + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        log.info("绑定删除成功！exchangeName="
                + binding.getExchangeName() + ", queueName=" + binding.getQueueName());
    }

    // 消息管理
    // 1. 新增消息
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        log.info("新增消息成功！messageId=" + message.getMessageId());
    }
    // 2. 根据 id 查询消息
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }
    // 3. 根据 id 删除消息
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        log.info("消息被移除！messageId=" + messageId);
    }
    // 4. 发送消息到指定队列 (将消息放入队列对应的链表中)
    public void sendMessage(MQueue queue, Message message) {
        // 1. 根据队列名，查找到该队列对应的消息链表，如果链表为空，那么创建一个链表（该队列的第一个消息）
        LinkedList<Message> messages
                = queueMessageMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>());
        // computeIfAbsent 这个方法是线程安全的

        // 2. 将消息加入对应队列的消息链表中
        // 加锁, 避免多个线程同时对这个 消息链表 进行操作
        synchronized (messages) {
            messages.add(message);
        }

        // 3. 把该消息加入到消息中心
        // 如果 message 已经存在于消息中心，重复插入也没关系
        // message 的内容是不会修改的，哈希表只是覆盖一下而已
        addMessage(message);
        log.info("消息被投递到队列中！messageId=" + message.getMessageId());
    }
    // 5. 从队列中获取消息 （从队列对应的消息链表中弹出消息）
    public Message pollMessage(String queueName) {
        // 1. 根据队列名，获取消息链表，如果为空，说明该队列中没有任何消息
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return null;
        }
        // 加锁，LinkedList 不是线程安全的
        synchronized (messages) {
            if (messages.size() == 0) {
                return null;
            }
            // 2. 取消息链表的头结点，进行头删
            Message currentMessage = messages.remove(0);
            log.info("消息从队列中取出！messageId=" + currentMessage.getMessageId());
            return currentMessage;
        }
    }
    // 6. 获取指定队列中的消息个数
    public int getMessageCount(String queueName) {
        // 1. 根据队列名，获取消息链表，如果为空，说明该队列中没有任何消息
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return 0;
        }
        synchronized (messages) {
            // 2. 返回消息链表大小
            return messages.size();
        }
    }

    // 待确认消息的管理
    // 1. 添加待确认消息 （消息被消费者取走后，加入待确认消息队列）
    public void addMessageWaitAck(String queueName, Message message) {
        // 1. 根据队列名，获取队列对应的消息表，如果没有，就新建一个表（该队列第一次插入待确认的消息）
        ConcurrentHashMap<String, Message> messageHashMap
                = queueMessageWaitAckMap.computeIfAbsent(queueName, k -> new ConcurrentHashMap<>());

        // 2. 插入未确认的消息 （ConcurrentHashMap 是线程安全的）
        messageHashMap .put(message.getMessageId(), message);
        log.info("消息进入待确认队列！messageId=" + message.getMessageId());
    }

    // 2. 删除未确认的消息（消息确认后调用）
    public void removeMessageWaitAck(String queueName, String messageId) {
        // 1. 根据队列名，获取队列对应的消息表
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null) {
            return;
        }
        // 2. 从链表中移除
        messageHashMap.remove(messageId);
        log.info("消息从待确认队列移除！messageId=" + messageId);
    }
    // 3. 获取指定的未确认的消息
    public Message getMessageWaitAck(String queueName, String messageId) {
        // 1. 根据队列名，获取队列对应的消息表
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null) {
            // 消息表不存在，返回 null
            return null;
        }
        // 2. 返回
        return messageHashMap.get(messageId);
    }


    // 从硬盘中读取数据，将数据从硬盘中恢复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MQException, ClassNotFoundException {
        // 1. 清空之前所有的数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        // 2. 恢复所有的交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for (Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(), exchange);
        }
        // 3. 恢复所有的队列数据
        List<MQueue> queues = diskDataCenter.selectAllQueues();
        for (MQueue queue : queues) {
            queueMap.put(queue.getName(), queue);
        }
        // 4. 恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings) {
            // 根据 exchangeName 查找对应的绑定表 bindingMap，如果没有就创建
            ConcurrentHashMap<String, Binding> bindingMap
                    = bindingsMap.computeIfAbsent(binding.getExchangeName(), k -> new ConcurrentHashMap<>());
            // 往当前的绑定表 bindingMap 中，添加绑定
            bindingMap.put(binding.getQueueName(), binding);
        }
        // 5. 恢复所有的消息数据
        // 消息是以队列的维度存储在文件中的
        // 所以要遍历所有的队列，恢复所有队列的消息
        for (MQueue queue : queues) {
            // 获取一个队列的所有消息
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            // 存入队列和消息的关联表
            queueMessageMap.put(queue.getName(), messages);
            for (Message message : messages) {
                // 存入消息中心
                messageMap.put(message.getMessageId(), message);
            }
        }
    }
}
